package cn.doitedu.rtmk.demo7;

import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.googlecode.aviator.AviatorEvaluator;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;


/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/12/25
 * @Desc: 学大数据，上多易教育

{
    "rule_id":"rule-D-001",
    "model_id":"D",
    "expressions":[
        "f1>f2 && f2>f3",
        "f1+f2>f3"
     ]
}


 **/
public class RuleMode_D_Calculator implements RuleCalculator{

    String ruleId;
    JSONObject message;
    JSONObject jsonObject;
    JSONArray expressions;
    HashMap<String, Object> dataEnv;


    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJson, Roaring64Bitmap preSelectUsersBitmap, Long histQueryEndTime) throws Exception {
        jsonObject = JSON.parseObject(ruleParamJson);
        ruleId = jsonObject.getString("rule_id");
        expressions = jsonObject.getJSONArray("expressions");

        message = new JSONObject();
        // {rule_id:rule-d-001,uid:1,hit_time:1237512358,expression_idx:1}
        message.put("rule_id",ruleId);


        dataEnv = new HashMap<>();
    }

    @Override
    public void calculate(UserEvent userEvent, Collector<String> out) throws IOException {

        Map<String, String> properties = userEvent.getProperties();
        int f1 = Integer.parseInt(properties.get("f1"));
        int f2 = Integer.parseInt(properties.get("f2"));
        int f3 = Integer.parseInt(properties.get("f3"));
        dataEnv.put("f1",f1);
        dataEnv.put("f2",f2);
        dataEnv.put("f3",f3);
        dataEnv.put("f4",f3);
        dataEnv.put("f5",f3);


        // 遍历表达式
        for(int i=0;i<expressions.size();i++){

            // "f1>f2 && f2>f3"
            String exp = expressions.getString(i);
            Boolean result = (Boolean) AviatorEvaluator.execute(exp, dataEnv);
            if(result){
                message.put("expression_idx",i);
                out.collect(message.toJSONString());
            }

        }


    }
}
